package com.zetyun.tiger.cdc.oracle;

import com.ververica.cdc.connectors.oracle.OracleSource;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.internal.DelegateSourceAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

@ActionMeta(
        id = "tiger.oracle-cdc",
        name = "Oracle CDC",
        tags = {"jdbc", "streaming"},
        category = "source",
        parameterClass = OracleCdcParameter.class,
        description = "参考自 https://github.com/ververica/flink-cdc-connectors/blob/master/docs/content/connectors/oracle-cdc.md "

)
@Slf4j
public class OracleCdcSourceAction extends DelegateSourceAction {

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        OracleCdcParameter param = context.getParameters();
//        log.info("{} received param: {}", getClass().getSimpleName(), param);

        SourceFunction<RtEvent> cdcSourceFunction = OracleSource.<RtEvent>builder()
                .hostname(param.getHostName())
                .port(param.getPort())
                .database(param.getDatabase()) // monitor XE database
                .schemaList(param.getSchemaList().toArray(new String[0])) // monitor inventory schema
                .tableList(param.getTableList().toArray(new String[0])) // monitor products table
                .username(param.getUserName())
                .password(param.getPassWord())
                .startupOptions(param.getStartupOptions())
                .debeziumProperties(param.toProp())
                .deserializer(new RtEventDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        try {
            delegate(cdcSourceFunction);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
